package com.cctc.rds.raft.core.client.command;

import com.cctc.rds.raft.core.RaftCluster;
import com.cctc.rds.raft.core.log.Log;
import com.cctc.rds.raft.core.log.LogService;
import com.cctc.rds.raft.core.log.commit.LogCommitter;
import com.cctc.rds.raft.core.log.commit.statemachine.StateMachine;
import com.cctc.rds.raft.core.node.Node;
import com.cctc.rds.raft.core.node.NodeRole;
import com.cctc.rds.raft.core.rpc.RpcClientRouter;
import com.cctc.rds.raft.core.rpc.election.ElectionListener;
import com.cctc.rds.raft.core.rpc.heartbeat.HeartBeatListener;
import com.cctc.rds.raft.core.rpc.heartbeat.HeartBeatRpcScheduler;
import com.cctc.rds.raft.core.rpc.log.LogRpc;
import com.cctc.rds.raft.core.rpc.log.LogRpcRequest;
import com.cctc.rds.raft.core.rpc.log.LogRpcScheduler;

import java.io.Closeable;
import java.io.IOException;
import java.util.Set;

public class RaftCommandClient implements ElectionListener, Closeable {
    private int nodeId;
    private LogService logService;
    private LogCommitter logApply;
    private LogRpcScheduler logRpcScheduler;
    private HeartBeatRpcScheduler heartBeatRpcScheduler;

    public RaftCommandClient(int nodeId, LogService logService, StateMachine stateMachine, RpcClientRouter<LogRpc> router) {
        this.nodeId = nodeId;
        this.logService = logService;
        this.logApply = new LogCommitter(logService, stateMachine);
        this.logRpcScheduler = new LogRpcScheduler(nodeId, logService, router);
        initCurTerm();
        seekCommitLogIndex();
    }

    private void initCurTerm() {
        Log log = logService.peek();
        long term = log != null ? log.getTerm() : -1;
        Node curNode = RaftCluster.getNode(nodeId);
        if (curNode != null) {
            curNode.initCurTerm(term);
        }
    }

    private void seekCommitLogIndex() {
        Node curNode = RaftCluster.getNode(nodeId);
        if (curNode != null) {
            curNode.seekCommitLogIndex(logApply.maxCommitLogIndex());
        }
    }

    @Override
    public void startElection() {
        heartBeatRpcScheduler.stop();
    }

    @Override
    public void endElection(NodeRole curRole) {
        heartBeatRpcScheduler.stop();
        if (curRole == NodeRole.Leader) {
            heartBeatRpcScheduler = new HeartBeatRpcScheduler(() -> {
                if (logRpcScheduler.heartbeat()) {
                    resetElectionRpcScheduler();
                }
            }, RaftCluster.getConfig().getHeartBeatMs());
        }
        heartBeatRpcScheduler.start();
    }

    private void resetElectionRpcScheduler() {
        try {
            HeartBeatListener listener = RaftCluster.getHeartBeatListener();
            if (listener != null) {
                listener.onHeartBeat();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public synchronized RaftCommandResponse exec(byte[] command) {
        RaftCommandResponse response = RaftCommandResponse.builder().build();
        Node curNode = RaftCluster.getNode(nodeId);
        Set<Node> nodes = RaftCluster.getAllNodes();
        if (curNode.getRole() != NodeRole.Leader) {
            for (Node node : nodes) {
                if (node.getRole() == NodeRole.Leader) {
                    response.setSuccess(false);
                    response.setRedirectToLeader(node.getIpPort());
                    return response;
                }
            }
            throw new RuntimeException("Raft 集群还未选举出 Leader 节点！！！");
        }
        Log log = Log.builder()
                .index(curNode.curCommitLogIndex() + 1)
                .term(curNode.getCurTerm())
                .command(command)
                .status(0)
                .build();
        logService.append(log);
        LogRpcRequest logRpcRequest = logRpcScheduler.newLogRpcRequest();
        logRpcRequest.setLogs(new Log[]{log});
        boolean flag = logRpcScheduler.append(logRpcRequest);
        response.setSuccess(flag);
        if (flag) {
            logApply.apply(log);
            curNode.seekCommitLogIndex(log.getIndex());
            logRpcScheduler.commit(log);
        }
        return response;
    }

    @Override
    public void close() throws IOException {
        logRpcScheduler.close();
        heartBeatRpcScheduler.stop();
    }
}
